package net.bwie.realtime.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.AverageSpeed;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.function.SpeedWindowFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import utils.KafkaUtil;

public class AverageSpeedMonitor {

    public static void main(String[] args) throws Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.读取数据源
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");

        //3.数据处理
        DataStream<String> resultStream = handle(carStream);
        resultStream.print();
        //4.数据输出

        //5.执行
        env.execute("AverageSpeedMonitor");
    }

    private static DataStream<String> handle(DataStream<String> stream) {

        SingleOutputStreamOperator<MonitorInfo> mapStream = stream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String value) throws Exception {
                return JSON.parseObject(value, MonitorInfo.class);
            }
        });

        // 窗口 滑动窗口，窗口长度是为5分钟，滑动步长为1分钟
        WindowedStream<MonitorInfo, String, TimeWindow> windowStream = mapStream.keyBy(v -> v.getMonitorId())
                .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)));

        //聚合计算
        SingleOutputStreamOperator<AverageSpeed> applyStream = windowStream.apply(new SpeedWindowFunction());

        SingleOutputStreamOperator<String> resultStream = applyStream.map(
                (MapFunction<AverageSpeed, String>) value -> JSON.toJSONString(value));
        return resultStream;
    }
}
